package com.juneyao.bigdata.service.impl;

import com.juneyao.bigdata.common.ConstantNV;
import com.juneyao.bigdata.domain.UserSendVo;
import com.juneyao.bigdata.service.HbaseSearchService;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

/**
 * @author quanbiao
 * @date 2018-06-28
 * hbase查询实现类
 */
@Service("hbaseSearchService")
public class HbaseSearchServiceImpl implements HbaseSearchService{

    private Logger logger = LoggerFactory.getLogger("HbaseSearchServiceImpl");

    @Autowired
    private Connection connection;

    /**
     * 短信标识是2
     */
    private static final String SMS_FLAG = "2";

    /**
     * 获取用户的详细细信息
     * @param userIds  用户id
     * @param qualifierStr 需要展示的列
     * @return
     */
    @Override
    public List<HashMap<String, String>> userDetailForShow(List<String> userIds, String qualifierStr) {
        //返回用户的详细信息集合
        List<HashMap<String, String>> retUserDetail = new ArrayList<>();
        try {
            //获取hbase表
            Table table = connection.getTable(TableName.valueOf(ConstantNV.HBASE_USER_DETAIL));

            //需要显示的列
            String[] qualifierArr = qualifierStr.split(",");

            //get集合
            List<Get> getList = this.getListByUserIdAndQualifier(userIds, Arrays.asList(qualifierArr));

            //查询结果
            Result[] results = table.get(getList);
            for (Result result : results) {
                if(!result.isEmpty()){
                    //获取每个用户需要展示的列和值
                    HashMap<String, String> userMap = new HashMap<>(16);
                    List<Cell> cells = result.listCells();
                    for (Cell cell : cells) {
                        userMap.put(Bytes.toString(CellUtil.cloneQualifier(cell)).toLowerCase(), Bytes.toString(CellUtil.cloneValue(cell)));
                    }
                    retUserDetail.add(userMap);
                }
            }
            logger.info("hbase search success");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return retUserDetail;
    }

    /**
     * 根据用户id和列组装get list查询
     * @param userIds
     * @param qualifierArr
     * @return
     */
    private List<Get> getListByUserIdAndQualifier(List<String> userIds, List<String> qualifierArr){
        //get批量查询
        List<Get> getList = new ArrayList<>();
        for (String userId : userIds) {
            //**********************************************

            //String hashUserId = String.format("%02d", Math.abs(userId.hashCode()%30))+"_"+userId;
            //  临时添加
            //**********************************************
            Get get = new Get(Bytes.toBytes(userId));
            //返回需要显示的列
            for (String qualifier : qualifierArr) {
                get.addColumn(Bytes.toBytes(ConstantNV.HBASE_FAMILY_NAME), Bytes.toBytes(qualifier.toUpperCase()));
            }
            //添加到get集合
            getList.add(get);
        }
        return getList;
    }

    /**
     * 批量获取用户发送的详细信息
     *
     * @param tableName
     * @param userIds
     * @param channel   渠道，短信取手机号，app和微信取会员信息
     * @return
     */
    @Override
    public List<UserSendVo> getUserSendDetail(String tableName, List<String> userIds, String channel) {
        List<UserSendVo> userSendVoList = new ArrayList<>();
        try {
            //获取hbase表
            Table table = connection.getTable(TableName.valueOf(tableName));

            //需要显示的列
            List<String> qualifierArr = new ArrayList<>();
            qualifierArr.add("PK_MEMBER_ID");
            qualifierArr.add("PK_CRM_INFO");
            //渠道(1:微信;2:短信；3：APP) 微信和app都带会员id，短信带手机号
            if(StringUtils.equals(channel, SMS_FLAG)){
                qualifierArr.add("MOBILE_PHONE");
            }

            //get list集合
            List<Get> getList = this.getListByUserIdAndQualifier(userIds, qualifierArr);

            //查询结果
            Result[] results = table.get(getList);
            for (Result result : results) {
                if(!result.isEmpty()){
                    //用户id
                    String userId = Bytes.toString(result.getRow());
                    //获取每个用户需要展示的列和值
                    UserSendVo userSendVo = new UserSendVo();
                    userSendVo.setUserId(userId);
                    List<Cell> cells = result.listCells();
                    for (Cell cell : cells) {
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = StringUtils.EMPTY;
                        if(null != CellUtil.cloneValue(cell)){
                            value = Bytes.toString(CellUtil.cloneValue(cell));
                        }
                        if(StringUtils.equals(channel, "1") || StringUtils.equals(channel, "3")){
                            if(StringUtils.equals(qualifier, "PK_MEMBER_ID")){
                                userSendVo.setPk_member_id(value);
                            }
                            userSendVo.setPk_crm_info(userId);
                            userSendVo.setMobile_phone(userId);

                        }else{
                            userSendVo.setPk_crm_info(value);
                            if(StringUtils.equals(qualifier, "PK_MEMBER_ID")){
                                userSendVo.setPk_member_id(value);
                            }
                            if(StringUtils.equals(qualifier, "MOBILE_PHONE")){
                                userSendVo.setMobile_phone(value);
                            }
                        }
                    }
                    userSendVoList.add(userSendVo);
                }
            }
            logger.info("hbase search success");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return userSendVoList;
    }

    /**
     * 根据rowKey和列获取值
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param qualifier
     * @return
     */
    @Override
    public Result getResultByrowkeyColumn(String tableName, String rowKey, String colFamily, String qualifier) {
        //获取hbase表
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(qualifier));
            Result result = table.get(get);
            return result;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取父活动，获取人数，使用逗号拼接，插入hbase
     *
     * @param parentActId
     * @param userIdsCompStr
     */
    @Override
    public void insertOneRow(String parentActId, String userIdsCompStr) {
        //获取hbase表
        try {
            Table table = connection.getTable(TableName.valueOf("tbl_act_pid_user_ids"));
            Put put = new Put(Bytes.toBytes(parentActId));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("userIds"), Bytes.toBytes(userIdsCompStr));
            table.put(put);
            logger.info("父活动id：" + parentActId + " 插入成功！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
